package com.study;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

/**
 * @author sjw
 * @description
 * @date 2019-08-19 上午11:43
 */
public class SourceRabbitMqSinkMySQL {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        //这些配置建议可以放在配置文件中，然后通过 parameterTool 来获取对应的参数值
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("localhost").setVirtualHost("/")
                .setPort(5672).setUserName("root").setPassword("123456")
                .build();

        SingleOutputStreamOperator<User> dataStreamSource = env.addSource(new RMQSource<>(connectionConfig,
                "q.study.flink",
                true,
                new SimpleStringSchema()))
                .setParallelism(1).map(new MapFunction<String, User>() {
                    @Override
                    public User map(String s) throws Exception {
                        return JsonHelper.jsonToObject(s, User.class);
                    }
                });

        //env.addSource()
        dataStreamSource.addSink(new SinkMySQL());


        //如果想保证 exactly-once 或 at-least-once 需要把 checkpoint 开启
//        env.enableCheckpointing(10000);
        env.execute("flink learning connectors rabbitmq");

    }
}
